… TaskEstimator (#564)

## Stack

1. #564
<- you are here
2. #567
3. #568
4. #569
5. #570

---

Introduces an event-driven API for reacting to distributed planning
lifecycle events, and replaces the previous TaskEstimator methods with
it.

## Motivation

- We need to extend existing methods with new capabilities without
imposing those changes on unrelated implementations. For example
#569.
- We need to add new event handlers without turning `TaskEstimator` into
a generic bucket for every planner lifecycle hook. For example
#570.
- We could be expressing current behavior in terms of just event handler
implementations, instead of inlining it in the code, for example:
#567.
- Converge into a single pattern all the user extension points, instead
of having different shape of APIs for them. For example:
#568

`TaskEstimator` was a convenient initial extension point, but its
methods run at different points in planning, observe different context,
and have different result semantics. Task-count estimation is evaluated
while traversing the plan; leaf specialization only runs after a stage's
final task count is known; routing runs later on the coordinator.
Keeping these operations in one trait couples otherwise independent
APIs.

This event-specific pattern is easier to extend without breaking API
consumers: a capability can evolve within the event that owns it instead
of requiring every `TaskEstimator` implementation to adopt a broader
trait. It is also easier to document and expose to users. Dedicated
event types give us focused API pages and code documentation, and
provide a quick index of the project’s extension points by the planner
lifecycle events where users can hook their code.

## What changed

Previously, one trait covered task-count estimation, leaf
specialization, and task routing:

```rust
trait TaskEstimator {
    fn task_estimation(&self, plan: &Arc<dyn ExecutionPlan>, cfg: &ConfigOptions)
        -> Option<TaskEstimation>;
    fn scale_up_leaf_node(&self, plan: &Arc<dyn ExecutionPlan>, task_count: usize,
        cfg: &ConfigOptions) -> Result<Option<Arc<dyn ExecutionPlan>>>;
    fn route_tasks(&self, ctx: &TaskRoutingContext<'_>)
        -> Result<Option<Vec<Url>>>;
}
```

The old API registered one implementation containing every hook:

```rust
SessionStateBuilder::new()
    .with_distributed_task_estimator(MyTaskEstimator);
```

Those phases now have dedicated event handlers, which can be ordinary
functions and are registered independently:

```rust
fn desired_task_count(
    _event: DesiredTaskCountEvent,
) -> Option<DesiredTaskCountEventResponse> {
    None
}

fn scale_up_leaf_node(
    _event: ScaleUpLeafNodeEvent,
) -> Option<Result<ScaleUpLeafNodeEventResponse>> {
    None
}

fn route_tasks(
    _event: RouteTasksEvent,
) -> Option<Result<RouteTasksEventResponse>> {
    None
}

SessionStateBuilder::new()
    .with_distributed_desired_task_count_handler(desired_task_count)
    .with_distributed_scale_up_leaf_node_handler(scale_up_leaf_node)
    .with_distributed_route_tasks_handler(route_tasks);
```

## Stack

1. #564
2. #567
<- you are here
3. #568
4. #569
5. #570

---

Moves routing policy into ordered `RouteTasksHandler`s. Built-ins now
cover child co-location, coordinator co-location, and randomized
fallback, while custom handlers can override them.

```rust
config.with_distributed_route_tasks_handler(|event| {
    Some(Ok(RouteTasksEventResponse::new(
        vec![worker_url.clone(); event.task_count],
    )))
});
```

---------

Co-authored-by: Jayant Shrivastava <jshrivastava03@gmail.com>